package com.ch.storm.demo;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import com.ch.storm.demo.bolt.WordCounter;
import com.ch.storm.demo.bolt.WordNormalizer;
import com.ch.storm.demo.spout.WordReader;

import java.util.HashMap;
import java.util.Map;

public class DemoTopology {

    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
        TopologyBuilder builder = new TopologyBuilder();

        Map<String, String> configMap = new HashMap<>();
        configMap.put("bootstrapServers", "10.144.101.40:9092,10.144.110.44:9092,10.134.83.54:9092");
        configMap.put("topic", "wordCount");
        configMap.put("groupId", "ch_test");

        //设置数据源
        builder.setSpout("spout", new WordReader(), 1);
        //读取spout数据源的数据，进行split业务逻辑
        builder.setBolt("split", new WordNormalizer(), 1).shuffleGrouping("spout");
        //读取split后的数据，进行count (tick周期10秒)
        builder.setBolt("count", new WordCounter(), 1).fieldsGrouping("split", new Fields("word"));
//        //读取count后的数据，进行缓冲打印 （tick周期3秒，仅仅为测试tick使用，所以多加了这个bolt）
//        builder.setBolt("show",new ShowBolt(),1).shuffleGrouping("count");
//        //读取show后缓冲后的数据，进行最终的打印 （实际应用中，最后一个阶段应该为持久层）
//        builder.setBolt("final",new FinalBolt(),1).allGrouping("show");

        Config config = new Config();
        config.setDebug(false);

        config.setNumWorkers(2);
        StormSubmitter.submitTopology(args[0], config, builder.createTopology());
    }
}
